package com.pw.study.flink.sql;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * @Author: linux_future
 * @since: 2022/3/12
 **/
public class $15SqlFunTable {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        //输出数据
        DataStreamSource<String> data = env.fromElements("hello,word,spark", "hello,flink,java,python", "docker,git");
        Table table = tEnv.fromDataStream(data, $("line"));

        mc1(table);
        mc2(table);
        mcSql(tEnv,table);
        mcSql2(tEnv,table);


    }

    private static void mcSql2(StreamTableEnvironment tEnv, Table table) {
        System.out.println("==================4.使用sql==");
        //注册方法，注册表
        tEnv.createTemporaryView("tb_word",table);
        tEnv.createTemporaryFunction("mySplit",MySplit.class);
        tEnv.sqlQuery("select line,word,len from tb_word join lateral table(mySplit(line)) on true ").execute().print();
        System.out.println("=====或者======");
        tEnv.sqlQuery("select line,word,len from tb_word , lateral table(mySplit(line)) ").execute().print();
        System.out.println("============== left join");
        tEnv.sqlQuery("select line,word,len from tb_word left join lateral table(mySplit(line)) on true ").execute().print();
        System.out.println("=====join或者left join给字段重命名======");
        tEnv.sqlQuery("select line,word1,len1 from tb_word left join lateral table(mySplit(line)) as T(word1,len1) on true ").execute().print();
    }


    private static void mcSql(StreamTableEnvironment tEnv, Table table) {
        System.out.println("==================3.使用sql==");
        tEnv.createTemporaryFunction("split",MySplit.class);
        table.joinLateral(call("split",$("line")))
                .select($("line"),$("word"),$("len")).execute().print();

    }

    private static void mc2(Table table) {
        System.out.println("=====2.左连接＋裂开");
        table.leftOuterJoinLateral(call(MySplit.class,$("line"))).select($("line"),$("word"),$("len")).execute().print();

    }

    private static void mc1(Table table) {
        System.out.println("====1. 内联使用====");
        table.joinLateral(call(MySplit.class,$("line"))).select("*").execute().print();
        table.joinLateral(call(MySplit.class,$("line")))
                .select($("line"),$("word"),$("len")).execute().print();

    }

    @FunctionHint(output = @DataTypeHint("ROW(word String,len int)"))
    public static class MySplit extends TableFunction<Row> {
        public void eval(String line) {
            if (line.length() == 0) {
                return;
            }
            for (String s : line.split(",")) {
                // 来一个字符串, 按照逗号分割, 得到多行, 每行为这个单词和他的长度
                collect(Row.of(s, s.length()));
            }
        }

    }
}
